package javaProduct;

import com.gizwits.util.CaseClasssDomain.User;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.Encoder;
import kafka.utils.VerifiableProperties;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;

public class KafkaProduct {
    private Producer<String, User> inner;

    private String brokerList;//for metadata discovery,spring setter
    private String location = "producer.properties";//spring setter

    private String defaultTopic;//spring setter

    public void setBrokerList(String brokerList) {
        this.brokerList = brokerList;
    }

    public void setLocation(String location) {
        this.location = location;
    }

    public void setDefaultTopic(String defaultTopic) {
        this.defaultTopic = defaultTopic;
    }

    public KafkaProduct(){
        try{
            init();

        }catch(Exception e) {
            e.printStackTrace();
        }
    }

    public void init() throws Exception {
        Properties properties = new Properties();
        properties.load(Thread.currentThread().getContextClassLoader().getResourceAsStream(location));


        if(brokerList != null) {
            properties.put("metadata.broker.list", brokerList);
        }
//        properties.put("metadata.broker.list", "s60:9092,s2:9092,s5:9092");
//        properties.put("serializer.class","javaProduct.KafkaProduct.UserEncode");
        ProducerConfig config = new ProducerConfig(properties);
        inner = new Producer(config);
    }

    public void send(User message){
        send(defaultTopic,message);
    }

    public void send(Collection<User> messages){
        send(defaultTopic, messages);
    }

    public void send(String topicName, User message) {
        if (topicName == null || message == null) {
            return;
        }
        KeyedMessage<String, User> km = new KeyedMessage(topicName,message);
        inner.send(km);
    }

    public void send(String topicName, Collection<User> users) {
        if (topicName == null || users == null) {
            return;
        }
        if (users.isEmpty()) {
            return;
        }
        List<KeyedMessage<String,User>> kms = new ArrayList();
        int i= 0;
        for (User entry : users) {
            KeyedMessage<String, User> km = new KeyedMessage(topicName,entry);
            kms.add(km);
            i++;
            if(i % 20 == 0){
                inner.send(kms);
                kms.clear();
            }
        }

        if(!kms.isEmpty()){
            System.out.println("------.......");
            inner.send(kms);
        }
    }

    public void close() {
        inner.close();
    }

    public  List genMsg(int num){

        ArrayList<User> users = new ArrayList<>();

        for (int i = 0; i <num; i++) {
              users.add(new User("feel"+i,100+i));
        }
         return users;
    }


    public static void main(String[] args) {
        KafkaProduct producer = null;
        try {
            producer = new KafkaProduct();
            List list = producer.genMsg(10);
            System.out.println(list);


            producer.send("testkafka-topic", list);



        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (producer != null) {
                producer.close();
            }
        }

    }

    public static class  UserEncode  implements Encoder<User> {
        public UserEncode(VerifiableProperties verifiableProperties) {
            /* This constructor must be present for successful compile. */
        }
        @Override
        public byte[] toBytes(User user) {
            ByteArrayOutputStream ba=null;
            try{
                ba = new ByteArrayOutputStream();
                ObjectOutputStream ob = new ObjectOutputStream(ba);
                ob.writeObject(user);

            }catch(Exception e)
            {
                e.printStackTrace();

            }
            return  ba.toByteArray();
        }
    }
}

